Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to Presto and Spark that use a high-performance format that works just like a SQL table.

官方的定义，iceberg是一种表格式。我们可以简单理解为他是基于计算层（flink、spark）和存储层（orc、parqurt）的一个中间层，我们可以把它定义成一种“数据组织格式”，Iceberg将其称之为“表格式”也是表达类似的含义。他与底层的存储格式（比如ORC、Parquet之类的列式存储格式）最大的区别是，它并不定义数据存储方式，而是定义了数据、元数据的组织方式，向上提供统一的“表”的语义。它构建在数据存储格式之上，其底层的数据存储仍然使用Parquet、ORC等进行存储。在hive建立一个iceberg格式的表。用flink或者spark写入iceberg，然后再通过其他方式来读取这个表，比如spark、flink、presto等。

Iceberg的架构和实现并未绑定于某一特定引擎，它实现了通用的数据组织格式，利用此格式可以方便地与不同引擎（如Flink、Hive、Spark）对接。

Apache Iceberg支持Apache Flink的DataStream Api和Table Api写记录进iceberg表。当前，我们只集成Iceberg和apache flink 1.11.x。

为了在flink中创建iceberg表，我们要求使用flink SQL client，因为这对使用者们来说更容易去理解概念。

准备两个jar包：

启动flink sql client，不带hive connector jar包，可以创建hadoop catalog如下：

启动flink sql client，带hive connector jar包，可以创建hadoop catalog和hive catalog如下：

　　注：

　　　　iceberg-flink-runtime-0.10.0.jar和flink-sql-connector-hive-2.2.0_2.11-1.11.2.jar不能放在flink安装目录的lib下，需要另外放置在别的目录

Flink1.11支持通过flink sql创建catalogs。catalog是Iceberg对表进行管理（create、drop、rename等）的一个组件。目前Iceberg主要支持HiveCatalog和HadoopCatalog两种Catalog。其中HiveCatalog将当前表metadata文件路径存储在hive Metastore，这个表metadata文件是所有读写Iceberg表的入口，所以每次读写Iceberg表都需要先从hive Metastore中取出对应的表metadata文件路径，然后再解析这个Metadata文件进行接下来的操作。而HadoopCatalog将当前表metadata文件路径记录在一个文件目录下，因此不需要连接hive Metastore。

创建一个名为hive_catalog的 iceberg catalog ，用来从 hive metastore 中加载表。

Iceberg 还支持 HDFS 中基于目录的 catalog ，可以使用’catalog-type’='hadoop’进行配置：

我们可以执行sql命令USE CATALOG hive_catalog来设置当前的catalog。

Flink也支持通过指定catalog-impl属性来加载自定义的Iceberg catalog接口。当catalog-impl设置了，catalog-type的值可以忽略，这里有个例子：

在启动SQL客户端之前，Catalogs可以通过在sql-client-defaults.yaml文件中注册。这里有个例子：

其中data目录下存储数据文件，metadata目录下存储元数据文件。

其中00001-62ade8ab-c1cf-40d3-bc21-fd5027bc3a55.metadata.json中存储表的shcema、partition spec以及当前snapshot manifests文件路径。snap-6771375506965563160-1-bb641961-162a-49a8-b567-885430d4e799.avro存储manifest文件路径。bb641961-162a-49a8-b567-885430d4e799-m0.avro记录本次提交的文件以及文件级别元数据。

Hadoopcatalog与Hivecatalog的data目录完全相同，metadata目录下文件稍有不同，HadoopCatalog管理的metadata目录如下所示：

说明该表的最新snapshot_id是2，即对应的snapshot元数据文件是v2.metadata.json，解析v2.metadata.json可以获取到该表当前最新snapshot对应的scheme、partition spec、父snapshot以及该snapshot对应的manifestList文件路径等，因此version-hint.text是HadoopCatalog获取当前snapshot版本的入口。

HiveCatalog的metadata目录下并没有version-hint.text文件，那它获取当前snapshot版本的入口在哪里呢？它的入口在Metastore中的schema里面，可以在HiveCatalog建表schema中的TBPROPERTIES中有个key是“metadata_location”，对应的value就是当前最新的snapshot文件。因此，有两点需要说明：

上面说到Iceberg目前支持两种Catalog，而且两种Catalog相互不兼容。那这里有两个问题：

社区是出于什么考虑实现两种不兼容的Catalog？

因为两者不兼容，必须选择其一作为系统唯一的Catalog，那是选择HiveCatalog还是HadoopCatalog，为什么？

先回答第一个问题。社区是出于什么考虑实现两种不兼容的Catalog？

在回答这个问题之前，首先回顾一下上一篇文章中介绍到的基于HadoopCatalog，Iceberg实现数据写入提交的ACID机制，最终的结论是使用了乐观锁机制和HDFS rename的原子性一起保障写入提交的ACID。如果某些文件系统比如S3不支持rename的原子性呢？那就需要另外一种机制保障写入提交的ACID，HiveCatalog就是另一种不依赖文件系统支持，但是可以提供ACID支持的方案，它在每次提交的时候都更新MySQL中同一行记录，这样的更新MySQL本身是可以保证ACID的。这就是社区为什么会支持两种不兼容Catalog的本质原因。

再来回答第二个问题。HadoopCatalog依赖于HDFS提供的rename原子性语义，而HiveCatalog不依赖于任何文件系统的rename原子性语义支持，因此基于HiveCatalog的表不仅可以支持HDFS，而且可以支持s3、oss等其他文件系统。但是HadoopCatalog可以认为只支持HDFS表，比较难以迁移到其他文件系统。但是HadoopCatalog写入提交的过程只依赖HDFS，不和Metastore/MySQL交互，而HiveCatalog每次提交都需要和Metastore/MySQL交互，可以认为是强依赖于Metastore，如果Metastore有异常，基于HiveCatalog的Iceberg表的写入和查询会有问题。相反，HadoopCatalog并不依赖于Metastore，即使Metastore有异常，也不影响Iceberg表的写入和查询。

考虑到我们目前主要还是依赖HDFS，同时不想强依赖于Metastore，所以我们选择HadoopCatalog作为我们系统唯一的Catalog。即使有一天，想要把HDFS上的表迁移到S3上去，也是可以办到的，大家想想，无论是HadoopCatalog还是HiveCatalog，数据文件和元数据文件本身都是相同的，只是标记当前最新的snapshot的入口不一样，那只需要简单的手动变换一下入口就可以实现Catalog的切换，切换到HiveCatalog上之后，就可以摆脱HDFS的依赖，问题并不大。

默认的，iceberg将会在flink中使用default数据库。如果我们不想在default数据库下面创建表，可以使用下面的例子去创建别的数据库。

注：default为数据库

表创建命令支持最常用的 flink create 子句，包括：

PARTITION BY (column1, column2, ...) 配置分区，apache flik 还不支持隐藏分区。

COMMENT 'table document'设置一个表描述。

WITH ('key'='value', ...)设置将存储在 apache iceberg 表属性中的表配置。

目前，它不支持计算列、主键和水印定义等。

要创建分区表，使用 PARTITIONED BY:

Apache Iceberg支持隐藏分区但apache flink不支持在列上按照函数分区，因此我们现在没有途径在flink DDL上支持隐藏分区，我们在未来将会改善flink DDL。

为了创建和另一张表具有相同结构、分区和表属性的一张表，使用CREATE TAABLE LIKE。

为了更详细，可以查看Flink CREATE TABLE documentation。

Iceberg 现在只支持在 flink 1.11中修改表属性。

Iceberg现在支持flink流式读和批量读。我们可以执行下面sql命令去把执行类型流式模式切换为批处理模式，如下：

如果在提交flink批处理作业时想要检查iceberg表中所有的记录，你可以执行下面的句子：

Iceberg支持处理flink流式作业中的增量数据，该数据从历史快照ID开始：

这些是可以在flink SQL提示选项中为流作业设置的选项：

现在Iceberg支持在flink1.11中使用insert into和insert overwrite。

flink 流作业将新数据追加到表中，使用 INSERT INTO:

要使用查询结果替换表中的数据，请在批作业中使用 INSERT OVERWRITE (flink 流作业不支持 INSERT OVERWRITE)。覆盖是 Iceberg 表的原子操作。

具有由 SELECT 查询生成的行的分区将被替换，例如:

Iceberg 还支持通过 select 值覆盖给定的分区:

对于分区的Iceberg表，当在PARTITION子句中为所有分区设置值时，它将插入到静态分区中；否则，如果在PARTITON子句中将部分分区列（所有分区列的前缀部分）设置为值，则将查询结果写入动态分区。对于未分区的Iceberg表，其数据将被INSERT OVERWRITE完全覆盖。

Iceberg现在支持使用Java API流式或者批量读取。

这个例子从Iceberg表读取所有记录，然后在flink批处理作业中打印到stdout控制台。

这个例子将会读取从快照id‘3821550127947089987’开始的增量记录，然后在flink流式作业中打印到stdout控制台中。

还有其他选项可以通过Java Api设置，详情请看FlinkSource#Builder.

Iceberg 支持从不同的 DataStream 输入写入 Iceberg 表。

我们支持在本地编写 DataStream < rowdata > 和 DataStream < Row> 到 sink iceberg 表。

为了动态覆盖现有 Iceberg 表中的数据，我们可以在FlinkSink构建器中设置overwrite标志。

现在Iceberg不支持在flink Sql中检查表，我们需要使用 iceberg’s Java API 去读取Iceberg来得到这些表信息。

Iceberg可以通过提交flink批作业去提供API重写小文件变为大文件。flink操作表现与spark的rewriteDataFiles.一样。

更多的重写文件操作选项文档，请看RewriteDataFilesAction

当前flink iceberg整合工作还有下面的特性不支持：

参考资料：

https://blog.csdn.net/u010834071/article/details/112507474  Flink结合Iceberg的一种实现方式笔记

https://zhengqiang.blog.csdn.net/article/details/112850376  Flink结合Kafka实时写入Iceberg实践笔记